// jfutureinterface.h included from jfuture.h
#include "precomp.h"
#include <qatomic.h>
#include "jthread.h"
#include "jthreadpool.h"
#include "private/jthreadpool_p.h"
#include "jfuture.h"
#include "jfutureinterface_p.h"

/**
 *
 */
enum {
    MaxProgressEmitsPerSecond = 25
};

/**
 * @brief JFutureInterfaceBase::JFutureInterfaceBase
 * @param initialState
 */
JFutureInterfaceBase::JFutureInterfaceBase(State initialState)
    : d(new JFutureInterfaceBasePrivate(initialState))
{ }

/**
 * @brief JFutureInterfaceBase::JFutureInterfaceBase
 * @param other
 */
JFutureInterfaceBase::JFutureInterfaceBase(const JFutureInterfaceBase &other)
    : d(other.d)
{
    d->refCount.ref();
}

/**
 * @brief JFutureInterfaceBase::~JFutureInterfaceBase
 */
JFutureInterfaceBase::~JFutureInterfaceBase()
{
    if (!d->refCount.deref())
        delete d;
}

/**
 * @brief JFutureInterfaceBase::cancel
 */
void JFutureInterfaceBase::cancel()
{
    QMutexLocker locker(d->m_mutex.mutex());
    if (d->state & Canceled)
        return;

    d->state = State((d->state & ~Paused) | Canceled);
    d->waitCondition.wakeAll();
    d->pausedWaitCondition.wakeAll();
    d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Canceled));
}

/**
 * @brief JFutureInterfaceBase::setPaused
 * @param paused
 */
void JFutureInterfaceBase::setPaused(bool paused)
{
    QMutexLocker locker(d->m_mutex.mutex());
    if (paused) {
        d->state = State(d->state | Paused);
        d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Paused));
    } else {
        d->state = State(d->state & ~Paused);
        d->pausedWaitCondition.wakeAll();
        d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Resumed));
    }
}

/**
 * @brief JFutureInterfaceBase::togglePaused
 */
void JFutureInterfaceBase::togglePaused()
{
    QMutexLocker locker(d->m_mutex.mutex());
    if (d->state & Paused) {
        d->state = State(d->state & ~Paused);
        d->pausedWaitCondition.wakeAll();
        d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Resumed));
    } else {
        d->state = State(d->state | Paused);
        d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Paused));
    }
}

/**
 * @brief JFutureInterfaceBase::setThrottled
 * @param enable
 */
void JFutureInterfaceBase::setThrottled(bool enable)
{
    // bail out if we are not changing the state
    if ((enable && (d->state & Throttled)) || (!enable && !(d->state & Throttled)))
        return;

    // lock and change the state
    QMutexLocker lock(d->m_mutex.mutex());
    if (enable) {
        d->state  = State(d->state | Throttled);
    } else {
        d->state  = State(d->state & ~Throttled);
        if (!(d->state & Paused))
            d->pausedWaitCondition.wakeAll();
    }
}

/**
 * @brief JFutureInterfaceBase::isRunning
 * @return
 */
bool JFutureInterfaceBase::isRunning() const
{
    return queryState(Running);
}

/**
 * @brief JFutureInterfaceBase::isStarted
 * @return
 */
bool JFutureInterfaceBase::isStarted() const
{
    return queryState(Started);
}

/**
 * @brief JFutureInterfaceBase::isCanceled
 * @return
 */
bool JFutureInterfaceBase::isCanceled() const
{
    return queryState(Canceled);
}

/**
 * @brief JFutureInterfaceBase::isFinished
 * @return
 */
bool JFutureInterfaceBase::isFinished() const
{
    return queryState(Finished);
}

/**
 * @brief JFutureInterfaceBase::isPaused
 * @return
 */
bool JFutureInterfaceBase::isPaused() const
{
    return queryState(Paused);
}

/**
 * @brief JFutureInterfaceBase::isThrottled
 * @return
 */
bool JFutureInterfaceBase::isThrottled() const
{
    return queryState(Throttled);
}

/**
 * @brief JFutureInterfaceBase::isResultReadyAt
 * @param index
 * @return
 */
bool JFutureInterfaceBase::isResultReadyAt(int index) const
{
    QMutexLocker lock(d->m_mutex.mutex());
    return d->internal_isResultReadyAt(index);
}

/**
 * @brief JFutureInterfaceBase::waitForNextResult
 * @return
 */
bool JFutureInterfaceBase::waitForNextResult()
{
    QMutexLocker lock(d->m_mutex.mutex());
    return d->internal_waitForNextResult();
}

/**
 * @brief JFutureInterfaceBase::waitForResume
 */
void JFutureInterfaceBase::waitForResume()
{
    // return early if possible to avoid taking the mutex lock.
    if ((d->state & Paused) == false || (d->state & Canceled))
        return;

    QMutexLocker lock(d->m_mutex.mutex());
    if ((d->state & Paused) == false || (d->state & Canceled))
        return;

    // decrease active thread count since this thread will wait.
    QThreadPool::globalInstance()->releaseThread();

    d->pausedWaitCondition.wait(d->m_mutex.mutex());

    QThreadPool::globalInstance()->reserveThread();
}

/**
 * @brief JFutureInterfaceBase::progressValue
 * @return
 */
int JFutureInterfaceBase::progressValue() const
{
    return d->m_progressValue;
}

/**
 * @brief JFutureInterfaceBase::progressMinimum
 * @return
 */
int JFutureInterfaceBase::progressMinimum() const
{
    return d->m_progressMinimum;
}

/**
 * @brief JFutureInterfaceBase::progressMaximum
 * @return
 */
int JFutureInterfaceBase::progressMaximum() const
{
    return d->m_progressMaximum;
}

/**
 * @brief JFutureInterfaceBase::resultCount
 * @return
 */
int JFutureInterfaceBase::resultCount() const
{
    QMutexLocker lock(d->m_mutex.mutex());
    return d->internal_resultCount();
}

/**
 * @brief JFutureInterfaceBase::progressText
 * @return
 */
jstring JFutureInterfaceBase::progressText() const
{
    QMutexLocker locker(d->m_mutex.mutex());
    return d->m_progressText;
}

/**
 * @brief JFutureInterfaceBase::isProgressUpdateNeeded
 * @return
 */
bool JFutureInterfaceBase::isProgressUpdateNeeded() const
{
    QMutexLocker locker(d->m_mutex.mutex());
    return !d->progressTime.isValid() || (d->progressTime.elapsed() > (1000 / MaxProgressEmitsPerSecond));
}

/**
 * @brief JFutureInterfaceBase::reportStarted
 */
void JFutureInterfaceBase::reportStarted()
{
    QMutexLocker locker(d->m_mutex.mutex());
    if ((d->state & Started) || (d->state & Canceled) || (d->state & Finished))
        return;

    d->setState(State(Started | Running));
    d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Started));
}

/**
 * @brief JFutureInterfaceBase::reportCanceled
 */
void JFutureInterfaceBase::reportCanceled()
{
    cancel();
}

/**
 * @brief JFutureInterfaceBase::reportException
 * @param exception
 */
void JFutureInterfaceBase::reportException(const JConcurrent::Exception &exception)
{
    QMutexLocker locker(d->m_mutex.mutex());
    if ((d->state & Canceled) || (d->state & Finished))
        return;

    d->m_exceptionStore.setException(exception);
    d->state = State(d->state | Canceled);
    d->waitCondition.wakeAll();
    d->pausedWaitCondition.wakeAll();
    d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Canceled));
}

/**
 * @brief JFutureInterfaceBase::reportFinished
 */
void JFutureInterfaceBase::reportFinished()
{
    QMutexLocker locker(d->m_mutex.mutex());
    if (!(d->state & Finished)) {
        d->state = State((d->state & ~Running) | Finished);
        d->waitCondition.wakeAll();
        d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Finished));
    }
}

/**
 * @brief JFutureInterfaceBase::setExpectedResultCount
 * @param resultCount
 */
void JFutureInterfaceBase::setExpectedResultCount(int resultCount)
{
    if (d->manualProgress == false)
        setProgressRange(0, resultCount);
    d->m_expectedResultCount = resultCount;
}

/**
 * @brief JFutureInterfaceBase::expectedResultCount
 * @return
 */
int JFutureInterfaceBase::expectedResultCount()
{
    return d->m_expectedResultCount;
}

/**
 * @brief JFutureInterfaceBase::queryState
 * @param state
 * @return
 */
bool JFutureInterfaceBase::queryState(State state) const
{
    return (d->state & state);
}

/**
 * @brief JFutureInterfaceBase::waitForResult
 * @param resultIndex
 */
void JFutureInterfaceBase::waitForResult(int resultIndex)
{
    d->m_exceptionStore.throwPossibleException();

    if (!(d->state & Running))
        return;

    // To avoid deadlocks and reduce the number of threads used, try to
    // run the runnable in the current thread.
    JThreadPool::globalInstance().d_func()->stealRunnable(d->runnable);

    QMutexLocker lock(d->m_mutex.mutex());

    if (!(d->state & Running))
        return;

    const int waitIndex = (resultIndex == -1) ? INT_MAX : resultIndex;
    while ((d->state & Running) && d->internal_isResultReadyAt(waitIndex) == false)
        d->waitCondition.wait(d->m_mutex.mutex());

    d->m_exceptionStore.throwPossibleException();
}

/**
 * @brief JFutureInterfaceBase::waitForFinished
 */
void JFutureInterfaceBase::waitForFinished()
{
    if (d->state & Running) {
        JThreadPool::globalInstance().d_func()->stealRunnable(d->runnable);

        QMutexLocker lock(d->m_mutex.mutex());

        while (d->state & Running)
            d->waitCondition.wait(d->m_mutex.mutex());
    }

    d->m_exceptionStore.throwPossibleException();
}

/**
 * @brief JFutureInterfaceBase::reportResultsReady
 * @param beginIndex
 * @param endIndex
 */
void JFutureInterfaceBase::reportResultsReady(int beginIndex, int endIndex)
{
    if ((d->state & Canceled) || (d->state & Finished) || beginIndex == endIndex)
        return;

    d->waitCondition.wakeAll();

    if (d->manualProgress == false) {
        if (d->internal_updateProgress(d->m_progressValue + endIndex - beginIndex) == false) {
            d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::ResultsReady,
                                               beginIndex,
                                               endIndex));
            return;
        }

        d->sendCallOuts(JFutureCallOutEvent(JFutureCallOutEvent::Progress,
                                            d->m_progressValue,
                                            d->m_progressText),
                        JFutureCallOutEvent(JFutureCallOutEvent::ResultsReady,
                                            beginIndex,
                                            endIndex));
        return;
    }
    d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::ResultsReady, beginIndex, endIndex));
}

/**
 * @brief JFutureInterfaceBase::setRunnable
 * @param runnable
 */
void JFutureInterfaceBase::setRunnable(JRunnable *runnable)
{
    d->runnable = runnable;
}

/**
 * @brief JFutureInterfaceBase::setFilterMode
 * @param enable
 */
void JFutureInterfaceBase::setFilterMode(bool enable)
{
    QMutexLocker locker(d->m_mutex.mutex());
    resultStoreBase().setFilterMode(enable);
}

/**
 * @brief JFutureInterfaceBase::setProgressRange
 * @param minimum
 * @param maximum
 */
void JFutureInterfaceBase::setProgressRange(int minimum, int maximum)
{
    QMutexLocker locker(d->m_mutex.mutex());
    d->m_progressMinimum = minimum;
    d->m_progressMaximum = maximum;
    d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::ProgressRange, minimum, maximum));
}

/**
 * @brief JFutureInterfaceBase::setProgressValue
 * @param progressValue
 */
void JFutureInterfaceBase::setProgressValue(int progressValue)
{
    setProgressValueAndText(progressValue, jstring());
}

/**
 * @brief JFutureInterfaceBase::setProgressValueAndText
 * @param progressValue
 * @param progressText
 */
void JFutureInterfaceBase::setProgressValueAndText(int progressValue,
                                                   const jstring &progressText)
{
    QMutexLocker locker(d->m_mutex.mutex());
    if (d->manualProgress == false)
        d->manualProgress = true;
    if (d->m_progressValue >= progressValue)
        return;

    if ((d->state & Canceled) || (d->state & Finished)) {
        return;
    }

    if (d->internal_updateProgress(progressValue, progressText)) {
        d->sendCallOut(JFutureCallOutEvent(JFutureCallOutEvent::Progress,
                                           d->m_progressValue,
                                           d->m_progressText));
    }
}

/**
 * @brief JFutureInterfaceBase::mutex
 * @return
 */
JMutex *JFutureInterfaceBase::mutex() const
{
    return &d->m_mutex;
}

/**
 * @brief JFutureInterfaceBase::exceptionStore
 * @return
 */
JConcurrent::internal::ExceptionStore &JFutureInterfaceBase::exceptionStore()
{
    return d->m_exceptionStore;
}

/**
 * @brief JFutureInterfaceBase::resultStoreBase
 * @return
 */
JConcurrent::ResultStoreBase &JFutureInterfaceBase::resultStoreBase()
{
    return d->m_results;
}

/**
 * @brief JFutureInterfaceBase::resultStoreBase
 * @return
 */
const JConcurrent::ResultStoreBase &JFutureInterfaceBase::resultStoreBase() const
{
    return d->m_results;
}

/**
 * @brief JFutureInterfaceBase::operator =
 * @param other
 * @return
 */
JFutureInterfaceBase &JFutureInterfaceBase::operator=(const JFutureInterfaceBase &other)
{
    other.d->refCount.ref();
    if (!d->refCount.deref())
        delete d;
    d = other.d;
    return *this;
}

/**
 * @brief JFutureInterfaceBase::referenceCountIsOne
 * @return
 */
bool JFutureInterfaceBase::referenceCountIsOne() const
{
    return d->refCount == 1;
}

/**
 * @brief JFutureInterfaceBasePrivate::JFutureInterfaceBasePrivate
 * @param initialState
 */
JFutureInterfaceBasePrivate::JFutureInterfaceBasePrivate(JFutureInterfaceBase::State initialState)
    : refCount(1), m_progressValue(0), m_progressMinimum(0), m_progressMaximum(0),
      state(initialState), pendingResults(0),
      manualProgress(false), m_expectedResultCount(0), runnable(0)//,
      //jmutex(&m_mutex)
{
    progressTime.invalidate();
}

/**
 * @brief JFutureInterfaceBasePrivate::internal_resultCount
 * @return
 */
int JFutureInterfaceBasePrivate::internal_resultCount() const
{
    return m_results.count(); // ### subtract canceled results.
}

/**
 * @brief JFutureInterfaceBasePrivate::internal_isResultReadyAt
 * @param index
 * @return
 */
bool JFutureInterfaceBasePrivate::internal_isResultReadyAt(int index) const
{
    return (m_results.contains(index));
}

/**
 * @brief JFutureInterfaceBasePrivate::internal_waitForNextResult
 * @return
 */
bool JFutureInterfaceBasePrivate::internal_waitForNextResult()
{
    if (m_results.hasNextResult())
        return true;

    while ((state & JFutureInterfaceBase::Running) && m_results.hasNextResult() == false)
        waitCondition.wait(m_mutex.mutex());

    return (!(state & JFutureInterfaceBase::Canceled) && m_results.hasNextResult());
}

/**
 * @brief JFutureInterfaceBasePrivate::internal_updateProgress
 * @param progress
 * @param progressText
 * @return
 */
bool JFutureInterfaceBasePrivate::internal_updateProgress(int progress,
                                                          const jstring &progressText)
{
    if (m_progressValue >= progress) {
        return false;
    }

    m_progressValue = progress;
    m_progressText = progressText;

    if (progressTime.isValid() && m_progressValue != m_progressMaximum) { // make sure the first and last steps are emitted.
        if (progressTime.elapsed() < (1000 / MaxProgressEmitsPerSecond)) {
            return false;
        }
    }

    progressTime.start();

    return true;
}

/**
 * @brief JFutureInterfaceBasePrivate::internal_setThrottled
 * @param enable
 */
void JFutureInterfaceBasePrivate::internal_setThrottled(bool enable)
{
    // bail out if we are not changing the state
    if ((enable && (state & JFutureInterfaceBase::Throttled))
            || (!enable && !(state & JFutureInterfaceBase::Throttled))) {
        return;
    }

    // change the state
    if (enable) {
        state  = JFutureInterfaceBase::State(state | JFutureInterfaceBase::Throttled);
    } else {
        state  = JFutureInterfaceBase::State(state & ~JFutureInterfaceBase::Throttled);
        if (!(state & JFutureInterfaceBase::Paused))
            pausedWaitCondition.wakeAll();
    }
}

/**
 * @brief JFutureInterfaceBasePrivate::sendCallOut
 * @param callOutEvent
 */
void JFutureInterfaceBasePrivate::sendCallOut(const JFutureCallOutEvent &callOutEvent)
{
    if (outputConnections.isEmpty()) {
        return;
    }

    for (int i = 0; i < outputConnections.count(); ++i) {
        outputConnections.at(i)->postCallOutEvent(callOutEvent);
    }
}

/**
 * @brief JFutureInterfaceBasePrivate::sendCallOuts
 * @param callOutEvent1
 * @param callOutEvent2
 */
void JFutureInterfaceBasePrivate::sendCallOuts(const JFutureCallOutEvent &callOutEvent1,
                                               const JFutureCallOutEvent &callOutEvent2)
{
    if (outputConnections.isEmpty()) {
        return;
    }

    for (int i = 0; i < outputConnections.count(); ++i) {
        JFutureCallOutInterface *_interface = outputConnections.at(i);
        _interface->postCallOutEvent(callOutEvent1);
        _interface->postCallOutEvent(callOutEvent2);
    }
}

// This function connects an output interface (for example a QFutureWatcher)
// to this future. While holding the lock we check the state and ready results
// and add the appropriate callouts to the queue. In order to avoid deadlocks,
// the actual callouts are made at the end while not holding the lock.

/**
 * @brief JFutureInterfaceBasePrivate::connectOutputInterface
 * @param _interface
 */
void JFutureInterfaceBasePrivate::connectOutputInterface(JFutureCallOutInterface *_interface)
{
    QMutexLocker locker(m_mutex.mutex());

    if (state & QFutureInterfaceBase::Started) {
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::Started));
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::ProgressRange,
                                                        m_progressMinimum,
                                                        m_progressMaximum));
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::Progress,
                                                        m_progressValue,
                                                        m_progressText));
    }

    JConcurrent::ResultIteratorBase it = m_results.begin();
    while (it != m_results.end()) {
        const int begin = it.resultIndex();
        const int end = begin + it.batchSize();
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::ResultsReady,
                                                        begin,
                                                        end));
        it.batchedAdvance();
    }

    if (state & QFutureInterfaceBase::Paused)
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::Paused));

    if (state & QFutureInterfaceBase::Canceled)
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::Canceled));

    if (state & QFutureInterfaceBase::Finished)
        _interface->postCallOutEvent(JFutureCallOutEvent(JFutureCallOutEvent::Finished));

    outputConnections.push_back(_interface);
}

/**
 * @brief JFutureInterfaceBasePrivate::disconnectOutputInterface
 * @param _interface
 */
void JFutureInterfaceBasePrivate::disconnectOutputInterface(JFutureCallOutInterface *_interface)
{
    QMutexLocker lock(m_mutex.mutex());
    const int index = outputConnections.indexOf(_interface);
    if (index == -1) {
        return;
    }
    outputConnections.removeAt(index);

    _interface->callOutInterfaceDisconnected();
}

/**
 * @brief JFutureInterfaceBasePrivate::setState
 * @param newState
 */
void JFutureInterfaceBasePrivate::setState(JFutureInterfaceBase::State newState)
{
    state = newState;
}
